package com.example.source;

import com.example.beans.Event;
import com.example.beans.SensorReading;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 *  
 *  * @projectName flink-1.13.2-study
 *  * @title     SensorSource   
 *  * @package    com.example.source  
 *  * @description    SenderSource 遥感器数据生成器  
 *  * @author hjx     
 *  * @date   2022-3-29 9:55  
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *  
 */
public class SensorSource implements SourceFunction<SensorReading> {
    // 声明一个布尔变量，作为控制数据生成的标识位
    private Boolean running = true;
    // 每隔多加生成一条数据
    private Long sleepTime = 1000L;
    // 生成多少条数据
    private int num = 100;

    public SensorSource() {
    }

    public SensorSource(Long sleepTime) {
        this.sleepTime = sleepTime;
    }

    public SensorSource(int num) {
        this.num = num;
    }

    public SensorSource(Long sleepTime, int num) {
        this.sleepTime = sleepTime;
        this.num = num;
    }

    @Override
    public void run(SourceContext<SensorReading> ctx) throws Exception {
        // 定义一个随机数发生器
        Random random = new Random();

        // 设置10个传感器的初始温度
        HashMap<String, Double> sensorTempMap = new HashMap<>();
        for (int i = 0; i <= num; i++) {
            sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
        }

        while (running) {
            for (String sensorId : sensorTempMap.keySet()) {
                // 在当前温度基础上随机波动
                Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                sensorTempMap.put(sensorId, newtemp);
                ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
            }
            // 控制输出频率
        }
    }


    @Override
    public void cancel() {
        running = false;
    }
}
